<?php
namespace App\Common\Lib\Com\RabbitMQ\_Channel\_Queue;

use App\Common as com;
use App\Common\Lib\Com\RabbitMQ\_Channel\_Queue\_Consume as name_Consume;
use App\Common\Lib\Com\RabbitMQ as nameRabbitMQ;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Exception\AMQPTimeoutException;

/**
 * @property nameRabbitMQ\DataStore $dataStore
 */
class _Consume extends com\Cla\ClaBase\CommonOfId\Client {

	private AMQPChannel $channel;

	/**
	 * @var string
	 */
	private string $queueName;

	/**
	 * 回调函数
	 * @var \Closure
	 */
	private \Closure $callback;

	/**
	 * 消费者标签
	 * @var string
	 */
	private string $consumerTag = '';

	/**
	 * 这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.
	 * @var bool
	 */
	private bool $noLocal = false;

	/**
	 * 收到消息后,是否不需要回复确认即被认为被消费
	 * @var bool
	 */
	private bool $noAck = false;

	/**
	 * 排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
	 * @var bool
	 */
	private bool $exclusive = false;

	/**
	 * 不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
	 * @var bool
	 */
	private bool $noWait = false;

	/**
	 *
	 * @var string
	 */
	private int|null $ticket = null;

	/**
	 * @var array
	 */
	private array $arguments = [];

	/**
	 * run事件回调
	 * @var \Closure[]
	 */
	private array $eventRunStartFns = [];

	/**
	 * run接收通知通道
	 * @var \Swoole\Coroutine\Channel
	 */
	private array $waitRunEndChannels = [];

	/**
	 * 运行状态
	 * @var int [0]未运行 [1]运行 [2]停止
	 */
	private int $runStatus = 0;

	/**
	 * @var int [0]无 [2]停止
	 */
	private int $runControl = 0;

	/**
	 * 运行控制
	 * @var int [0]无 [2]停止
	 */
	// private int $runControl = 0;

	private ?\Swoole\Coroutine\Channel $runStatusChannalStop = null;

	/**
	 * _Exchange constructor.
	 * @param com\Cla\ClaBase\DataStore $dataStore
	 * @param AMQPChannel $channel
	 * @param string $queueName
	 * @throws
	 */
	public function __construct( com\Cla\ClaBase\DataStore $dataStore, AMQPChannel $channel, string $queueName ) {
		parent::__construct( $dataStore );
		//
		$this->channel = $channel;
		//
		$this->queueName = $queueName;
	}

	/**
	 * 回调函数
	 * @param \Closure $callback
	 * @return $this
	 */
	public function callback( \Closure $callback ):self {
		$this->callback = function(  ...$args ) use ( $callback ) {
			if( $this->isRunStatus() ){
				$callback( ...$args );
			}
		};
		//
		return $this;
	}

	/**
	 * 消费者标签
	 * @param string $consumerTag
	 * @return $this
	 */
	public function consumerTag( string $consumerTag = '' ):self {
		$this->consumerTag = $consumerTag;
		//
		return $this;
	}

	/**
	 * 这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.
	 * @param bool $noLocal
	 * @return $this
	 */
	public function noLocal( bool $noLocal ):self {
		$this->noLocal = $noLocal;
		//
		return $this;
	}

	/**
	 * 收到消息后,是否不需要回复确认即被认为被消费
	 * @param bool $noAck
	 * @return $this
	 */
	public function noAck( bool $noAck ):self {
		$this->noAck = $noAck;
		//
		return $this;
	}

	/**
	 * 排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
	 * @param bool $exclusive
	 * @return $this
	 */
	public function exclusive( bool $exclusive ):self {
		$this->exclusive = $exclusive;
		//
		return $this;
	}

	/**
	 * 不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
	 * @param bool $noWait
	 * @return $this
	 */
	public function noWait( bool $noWait ):self {
		$this->noWait = $noWait;
		//
		return $this;
	}

	/**
	 * @param int|null $ticket
	 * @return $this
	 */
	public function ticket( int|null $ticket = null ):self {
		$this->ticket = $ticket;
		//
		return $this;
	}

	/**
	 * @param array $arguments
	 * @return $this
	 */
	public function arguments( array $arguments = [] ):self {
		$this->arguments = $arguments;
		//
		return $this;
	}

	/**
	 * run事件回调
	 * @param \Closure|null $eventFn
	 * @param string $fnSid 回调sid，如果已存在则覆盖
	 * @return $this
	 */
	public function setEventRunStart( ?\Closure $eventFn = null, string $fnSid = null ):self {
		if( $fnSid === null ){
			$this->eventRunStartFns[] = $eventFn;
		}else{
			$this->eventRunStartFns[ $fnSid ] = $eventFn;
		}

		//
		return $this;
	}

	/**
	 * 是否存在指定的run事件回调
	 * @param string $fnSid 回调sid，如果已存在则覆盖
	 * @return $this
	 */
	public function hasEventRunStart( string $fnSid = null ):bool {
		return array_key_exists( $fnSid, $this->eventRunStartFns );
	}

	/**
	 * 协程运行
	 * @return $this
	 */
	public function runOfCoroutine(  ):self {
		$channel = new \Swoole\Coroutine\Channel();
		//
		$this->setEventRunStart( function() use ( $channel ) {
			$channel->push( true );
		} );
		//
		go( function(){
			$this->run();
			//
			foreach( $this->waitRunEndChannels as $waitRunEndChannel ){
				$waitRunEndChannel->push( true );
			}
			// d('runOfCoroutine->go->end');
		} );
		//
		$channel->pop();
		//
		return $this;
	}

	/**
	 * 等待run协程运行
	 * @param \Closure $waitingFn 等待中的回调
	 * @return $this
	 */
	public function waitRunEndOfCoroutine( \Closure $waitingFn = null ):self {
		//
		if( ! $this->isRunStatus() ){
			return $this;
		}
		//
		$channel = new \Swoole\Coroutine\Channel();
		//
		$this->waitRunEndChannels[] = $channel;
		//
		if( $waitingFn instanceof \Closure ){
			call_user_func( $waitingFn );
		}
		//
		$channel->pop();
		//
		return $this;
	}

	/**
	 * 运行
	 * @return $this
	 * @throws
	 */
	public function run():self {
		$this->runStatus = 1;
		//
		$this->consumerTag = $this->channel->basic_consume(
			$this->queueName,
			$this->consumerTag,
			$this->noLocal,
			$this->noAck,
			$this->exclusive,
			$this->noWait,
			$this->callback,
			$this->ticket
		);
		// d('basic_consume');
		// 执行回调
		foreach( $this->eventRunStartFns as $eventRunStartFn ){
			if( is_callable( $eventRunStartFn ) ){
				call_user_func( $eventRunStartFn );
			}
		}
		//
		// d('开启接收器');
		while( $this->runStatus === 1 && $this->channel->is_consuming() ){
			try{
				$this->channel->wait( null, false, 1 );
			}catch( AMQPTimeoutException $e ){
			}catch( \PhpAmqpLib\Exception\AMQPInvalidFrameException $e ){
				echo 'ERROR:1';
			}
			//
			if( $this->runControl === 2 ){
				$this->runControl = 0; // 控制复位
				$this->channel->basic_cancel( $this->consumerTag );
			}
		}
		//
		$this->runStatus = 0;
		// $this->channel->close();
		$this->callback = function(  ...$args ){};
		// d('已退出接收器');
		//
		return $this;
	}

	/**
	 * 停止
	 * @return $this
	 */
	public function stop(  ):self {
		if( $this->runStatus === 1 ){
			$this->runControl = 2;
		}
		//
		return $this;
	}

	/**
	 * 停止并等待停止
	 * @return $this
	 */
	public function stopAndWait(  ):self {
		// d('停止并等待停止...1');
		if( $this->runStatus === 1 ){
			// d('停止并等待停止...2');
			//
			$this->waitRunEndOfCoroutine( function(){
				// d('停止并等待停止...3');
				$this->runControl = 2;
			} );
		}
		// d('停止并等待停止...4');
		//
		return $this;
	}

	/**
	 * 是否为运行中
	 * @return bool 是否为运行中
	 */
	public function isRunStatus(  ):bool {
		return $this->runStatus === 1;
	}
	
}
